package com.hr.reactive.flux;

import com.hr.reactive.common.Utils;
import reactor.core.publisher.Flux;

import java.util.concurrent.atomic.AtomicInteger;

public class Lec12FluxGenerateState {
    public static void main(String[] args) {
        demo1();
    }

    private static void demo() {
        AtomicInteger counter = new AtomicInteger(0);
        //generate无状态  循环执行
        Flux.generate(synchronousSink -> {
            var name = Utils.faker().country().name();
            synchronousSink.next(name);
            counter.getAndIncrement();
            System.out.println(Thread.currentThread().getName());
            if (counter.get() == 10 || name.equals("Canada")) {
                System.out.println(counter);
                System.out.println(name);
                synchronousSink.complete();
            }
        }).subscribe(Utils.subscriber());
    }

    private static void demo1() {
        //generate无状态  循环执行
        Flux.generate(() -> 0, (counter, synchronousSink) -> {
            var name = Utils.faker().country().name();
            synchronousSink.next(name);
            counter++;
            if (counter == 10 || name.equals("Canada")) {
                System.out.println(counter);
                System.out.println(name);
                synchronousSink.complete();
            }
            return counter;
        }).subscribe(Utils.subscriber());
    }
}
